package com.lam.framework.async;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.lam.common.components.redis.RedisUtil;
import com.lam.common.utils.CollectionUtil;
import com.lam.common.utils.IpUtils;
import com.lam.common.utils.SpringUtils;
import com.lam.framework.async.model.AbstractRunTask;
import com.lam.framework.async.model.CallableTask;
import com.lam.framework.async.model.RunnableTask;

/**
 * 异步任务管理器
 * 
 * @author lam
 */
@Component
public class AsyncManager {
	
	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	
	/** 操作延迟10毫秒 */
	private static final int OPERATE_DELAY_TIME = 10;
	/** 异步操作任务调度线程池 */
	@Resource(name="scheduledExecutorService")
	private ScheduledExecutorService executor;

	/** 单例模式 */
	private static AsyncManager instance;
	
	private final Map<String, Future<?>> taskMap = new HashMap<>();

	public static AsyncManager me() {
		if(instance == null) {
			instance = SpringUtils.getBean(AsyncManager.class);
		}
		return instance;
	}
	
	public static String getExecutorName() {
		return "LamSchedulePool";
	}
	
	public static void init() {
		Set<String> keys = RedisUtil.keys("asyncTask:key:");
		if(CollectionUtil.isEmpty(keys)) {
			return;
		}
		String curIp = IpUtils.getHostIp();
		for (String key : keys) {
			String val = (String) RedisUtil.get(key);
			if(val.contains(curIp)) {
				RedisUtil.del(key);
			}
		}
	}

	/**
	 * 将该线程的执行状态置为已完成
	 * @param runTask
	 */
	public static void done(AbstractRunTask<?> runTask) {
		Future<?> future = me().taskMap.get(runTask.getKey());
		if(future != null) {
			me().taskMap.remove(runTask.getKey());
			RedisUtil.del(taskRedisKey(runTask.getKey()));
		}		
	}
	
	/**
	 * 查询指定线程key的线程是否正在异步执行中
	 * @param key
	 * @return
	 */
	public static boolean isRunning(String key) {
		return me().taskMap.containsKey(key) || RedisUtil.hasKey(taskRedisKey(key));
	}
	
	public static String getRunInfo(String key) {
		
		return isRunning(key)? (String) RedisUtil.get(taskRedisKey(key)): null;
	}
	
	private static String taskRedisKey(String key) {
		return String.format("asyncTask:key:%s", key);
	}
	
	/**
	 * 执行任务，添加到线程表，跨线程可以获取该执行任务的线程状态
	 * @param task 任务
	 */
	public static Future<?> execute(String key, Runnable task) {
		Future<?> future = execute(new RunnableTask(key, task));
		me().taskMap.put(key, future);
		RedisUtil.set(taskRedisKey(key), String.format("%s-%s.%s", IpUtils.getHostIp(), IpUtils.getHostName(), IdWorker.getId()));
		return future;
	}
	
	/**
	 * 执行任务（任务有返回值），添加到线程表，跨线程可以获取该执行任务的线程状态
	 * @param task 任务
	 */
	public static <T> Future<T> execute(String key, Callable<T> task) {
		Future<T> future = execute(new CallableTask<T>(key, task));
		me().taskMap.put(key, future);
		RedisUtil.set(taskRedisKey(key), String.format("%s-%s.%s", IpUtils.getHostIp(), IpUtils.getHostName(), IdWorker.getId()));
		return future;
	}
	
	/**
	 * 执行任务，不添加到线程表，跨线程不能获取该执行任务的线程状态
	 * @param task 任务
	 */
	public static Future<?> execute(Runnable task) {
		return me().executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 执行任务（任务有返回值），不添加到线程表，跨线程不能获取该执行任务的线程状态
	 * @param task 任务
	 */
	public static <T> Future<T> execute(Callable<T> task) {
		return me().executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
	}

	/**
	 * 停止任务线程池
	 */
	public static void shutdown() {
		ThreadUtils.shutdownAndAwaitTermination(me().executor);
	}
	
	@PreDestroy
	public void destroy() {
		try {
			logger.info("====关闭后台异步任务线程池！====");
			AsyncManager.shutdown();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

}
